//
// Copyright (c) 2014, 保留所有权利。
//
// TableSync.java
//
// 作    者：王红燕
// 当前版本：1.0
// 2014 上午10:36:03
//
// 
//
package com.vanke.agent.sync.copy;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashSet;
import java.util.Set;

import org.apache.log4j.Logger;

import com.vanke.agent.ADataSource;
import com.vanke.agent.common.CstUtil;
import com.vanke.agent.sync.AgentSync;

public class TableSync extends AgentSync {
	protected Logger log = null;
	protected Set<String> aready = new HashSet<String>();

	@Override
	public void sync() {
		if (log == null) {
			log = getLog();
		}
		String table = getTable();
		String fields = getFields();
		String guidField = getGuidField();
		if (table == null || fields == null || guidField == null) {
			log.warn("缺少必须数据，不执行同步");
			return;
		} //~

		aready.clear();
		Connection bizCon = ADataSource.getConnection(ADataSource.SCHEMA_BIZ);
		try {
			loadGUID(bizCon);
			syncTable(bizCon, table, fields, guidField);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			aready.clear();
			ADataSource.close(bizCon);
		}
	} //~sync

	protected String getTable() {
		return null;
	}

	protected String getFields() {
		return null;
	}

	protected String getGuidField() {
		return null;
	}

	protected Logger getLog() {
		return Logger.getLogger(TableSync.class);
	}

	/* 缓存已有的GUID */
	protected void loadGUID(Connection con) throws SQLException {
		Statement stmt = null;
		ResultSet rs = null;
		String table = getTable();
		String guidField = getGuidField();
		try {
			stmt = con.createStatement();
			rs = stmt.executeQuery("select " + guidField + " from " + table);
			while (rs.next()) {
				aready.add(rs.getString(1));
			}
		} catch (SQLException e) {
			e.printStackTrace();
			throw e;
		} finally {
			ADataSource.close(rs, stmt, null);
		}
	} //~

	protected void syncTable(Connection bizCon, String table, String fields,
			String guidField) {
		Connection myCon = ADataSource.getConnection(ADataSource.SCHEMA_MY);
		long now = System.currentTimeMillis();
		int add = 0, update = 0;
		log.info("SYNC_BEGIN[" + table + "]");

		PreparedStatement queryPstmt = null;
		ResultSet queryRs = null;

		PreparedStatement addPstmt = null;
		PreparedStatement updatePstmt = null;
		try {
			bizCon.setAutoCommit(false);
			// 新增
			addPstmt = bizCon.prepareStatement(CstUtil.formatInsertSql(table,
					fields));

			// 更新
			String updateSql = CstUtil.formatUpdateSql(table, fields);
			updateSql += " where " + guidField + " = ?";
			updatePstmt = bizCon.prepareStatement(updateSql);

			// 查询
			String sql = "select " + fields + " from " + table;
			queryPstmt = myCon.prepareStatement(sql);
			queryRs = queryPstmt.executeQuery();

			// 遍历
			int columns = fields.split(",").length;
			while (queryRs.next()) {
				String guid = queryRs.getString(guidField);
				if (aready.contains(guid)) {
					update++;
					for (int i = 1; i <= columns; i++) {
						updatePstmt.setString(i, queryRs.getString(i));
					}
					updatePstmt.setString(columns + 1, guid);
					updatePstmt.addBatch();
				} else {
					add++;
					for (int i = 1; i <= columns; i++) {
						addPstmt.setString(i, queryRs.getString(i));
					}
					addPstmt.addBatch();
				} // end if

				// 无论batchSize是多少，最多一次提交不能超过10000
				if ((add + update) % 2000 == 0) {
					addPstmt.executeBatch();
					updatePstmt.executeBatch();
					bizCon.commit();
					addPstmt.clearBatch();
					updatePstmt.clearBatch();
					log.info(add + " added and " + update + " updated, spend "
							+ (System.currentTimeMillis() - now) + "ms");
				} // end if
			} // end while
			addPstmt.executeBatch();
			updatePstmt.executeBatch();
			bizCon.commit();
			addPstmt.clearBatch();
			updatePstmt.clearBatch();

			now = System.currentTimeMillis() - now;
			log.info(table + " all " + add + " added and " + update
					+ " updated, spend" + now + "ms");
		} catch (Exception e) {
			e.printStackTrace();
			log.error("同步表" + table + "异常：" + e);
		} finally {
			ADataSource.close(queryRs, queryPstmt, myCon);
			ADataSource.close(addPstmt);
			ADataSource.close(updatePstmt);
		}
		log.info("SYNC_END[" + table + "]");
	} //~syncTable
}
